package com.test.reactor_test;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

/**
 * @author zxs
 * @version 1.0
 * @date 2019-03-18
 */
public class ArraySubscriber<T> implements Subscription {

    private final Subscriber< ? super T> actual;

    private final T[] data;

    private int index;

    private boolean cancel;

    public ArraySubscriber(Subscriber<? super T> actual, T[] data) {
        this.actual = actual;
        this.data = data;
    }

    @Override
    public void request(long l) {
        if(cancel){
            return;
        }
        long length = data.length;
        for(int i=0;i<l && index<length;i++){
            actual.onNext(data[index++]);
            if(index==length){
                actual.onComplete();
            }
        }
    }

    @Override
    public void cancel() {
        this.cancel=true;
    }
}
